/*-------------------------------------------------------------------------
 *
 * nodeMergeAppend.c
 *	  处理MergeAppend节点的例程。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/executor/nodeMergeAppend.c
 *
 *-------------------------------------------------------------------------
 */
/* 接口例程
 *		ExecInitMergeAppend		- 初始化 MergeAppend 节点
 *		ExecMergeAppend			- 从节点中检索下一个元组
 *		ExecEndMergeAppend		- 关闭 MergeAppend 节点
 *		ExecReScanMergeAppend	- 重新扫描 MergeAppend 节点
 *
 *	 备注
 *		A MergeAppend 节点包含一个或多个子计划的列表。
 *		这些子计划预计会根据共同的排序键提供已经排序的元组。
 *		MergeAppend 节点合并这些流
 *		以生成以相同方式排序的输出。
 *
 *		MergeAppend 节点不使用其左右
 *		子树，而是维护一个子计划列表，因此
 *		一个典型的 MergeAppend 节点在计划树中的样子如下：
 *
 *				   ...
 *				   /
 *				MergeAppend---+------+------+--- nil
 *				/	\		  |		 |		|
 *			  nil	nil		 ...    ...    ...
 *								 子计划
 */

#include "postgres.h"

#include "executor/execdebug.h"
#include "executor/execPartition.h"
#include "executor/nodeMergeAppend.h"
#include "lib/binaryheap.h"
#include "miscadmin.h"

/*我们为堆数组中的每个项目拥有一个槽。我们使用SlotNumber来存储槽索引。这实际上并不提供任何正式的类型安全，但使代码更具自文档化。*/
typedef int32 SlotNumber;

static TupleTableSlot *fc_ExecMergeAppend(PlanState *fc_pstate);
static int	fc_heap_compare_slots(Datum fc_a, Datum fc_b, void *fc_arg);


/* ----------------------------------------------------------------
 *		ExecInitMergeAppend
 *
 *		开始 MergeAppend 节点的所有子扫描。
 * ----------------------------------------------------------------
 */
MergeAppendState *
ExecInitMergeAppend(MergeAppend *fc_node, EState *fc_estate, int fc_eflags)
{
	MergeAppendState *fc_mergestate = makeNode(MergeAppendState);
	PlanState **fc_mergeplanstates;
	Bitmapset  *fc_validsubplans;
	int			fc_nplans;
	int			fc_i,
				fc_j;

	/* 检查不支持的标志 */
	Assert(!(fc_eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));

	/*
	 * 为我们的节点创建新的 MergeAppendState
	 */
	fc_mergestate->ps.plan = (Plan *) fc_node;
	fc_mergestate->ps.state = fc_estate;
	fc_mergestate->ps.ExecProcNode = fc_ExecMergeAppend;

	/* 如果启用了运行时分区修剪，请立即设置该项 */
	if (fc_node->part_prune_info != NULL)
	{
		PartitionPruneState *fc_prunestate;

		/*
		 * 设置修剪数据结构。这还通过考虑执行初始修剪的结果（如果有）来初始化有效子计划的集合。
		 */
		fc_prunestate = ExecInitPartitionPruning(&fc_mergestate->ps,
											  list_length(fc_node->mergeplans),
											  fc_node->part_prune_info,
											  &fc_validsubplans);
		fc_mergestate->ms_prune_state = fc_prunestate;
		fc_nplans = bms_num_members(fc_validsubplans);

		/*
		 * 当不需要运行时修剪且至少有一个
		 * 子计划时，我们可以立即填充 ms_valid_subplans，防止
		 * 后续调用 ExecFindMatchingSubPlans。
		 */
		if (!fc_prunestate->do_exec_prune && fc_nplans > 0)
			fc_mergestate->ms_valid_subplans = bms_add_range(NULL, 0, fc_nplans - 1);
	}
	else
	{
		fc_nplans = list_length(fc_node->mergeplans);

		/*
		 * 当未启用运行时分区修剪时，我们可以将所有子计划标记为有效；它们也必须全部初始化。
		 */
		Assert(fc_nplans > 0);
		fc_mergestate->ms_valid_subplans = fc_validsubplans =
			bms_add_range(NULL, 0, fc_nplans - 1);
		fc_mergestate->ms_prune_state = NULL;
	}

	fc_mergeplanstates = (PlanState **) palloc(fc_nplans * sizeof(PlanState *));
	fc_mergestate->mergeplans = fc_mergeplanstates;
	fc_mergestate->ms_nplans = fc_nplans;

	fc_mergestate->ms_slots = (TupleTableSlot **) palloc0(sizeof(TupleTableSlot *) * fc_nplans);
	fc_mergestate->ms_heap = binaryheap_allocate(fc_nplans, fc_heap_compare_slots,
											  fc_mergestate);

	/*
	 * 杂项初始化
	 *
	 * MergeAppend 节点具有 Result 插槽，用于保存元组的指针，
	 * 所以我们必须对它们进行初始化。 FIXME
	 */
	ExecInitResultTupleSlotTL(&fc_mergestate->ps, &TTSOpsVirtual);

	/* 节点从其每个子节点返回槽，因此不固定 */
	fc_mergestate->ps.resultopsset = true;
	fc_mergestate->ps.resultopsfixed = false;

	/*
	 * 调用 ExecInitNode 对每个有效计划进行执行，并将
	 * 结果保存到 mergeplanstates 数组中。
	 */
	fc_j = 0;
	fc_i = -1;
	while ((fc_i = bms_next_member(fc_validsubplans, fc_i)) >= 0)
	{
		Plan	   *fc_initNode = (Plan *) list_nth(fc_node->mergeplans, fc_i);

		fc_mergeplanstates[fc_j++] = ExecInitNode(fc_initNode, fc_estate, fc_eflags);
	}

	fc_mergestate->ps.ps_ProjInfo = NULL;

	/*
	 * 初始化排序键信息
	 */
	fc_mergestate->ms_nkeys = fc_node->numCols;
	fc_mergestate->ms_sortkeys = palloc0(sizeof(SortSupportData) * fc_node->numCols);

	for (fc_i = 0; fc_i < fc_node->numCols; fc_i++)
	{
		SortSupport fc_sortKey = fc_mergestate->ms_sortkeys + fc_i;

		fc_sortKey->ssup_cxt = CurrentMemoryContext;
		fc_sortKey->ssup_collation = fc_node->collations[fc_i];
		fc_sortKey->ssup_nulls_first = fc_node->nullsFirst[fc_i];
		fc_sortKey->ssup_attno = fc_node->sortColIdx[fc_i];

		/*
		 * 执行简化的键转换是不可行的，因为
		 * 元组在需要时被提取到合并状态的二进制堆中。
		 * 在提取时将元组转换为简化表示形式很可能适得其反，
		 * 所以完全放弃这种额外的优化。
		 */
		fc_sortKey->abbreviate = false;

		PrepareSortSupportFromOrderingOp(fc_node->sortOperators[fc_i], fc_sortKey);
	}

	/*
	 * 初始化以显示我们尚未运行子计划
	 */
	fc_mergestate->ms_initialized = false;

	return fc_mergestate;
}

/* ----------------------------------------------------------------
 *	   ExecMergeAppend
 *
 *		处理多个子计划的迭代。
 * ----------------------------------------------------------------
 */
static TupleTableSlot * fc_ExecMergeAppend(PlanState *fc_pstate)
{
	MergeAppendState *fc_node = castNode(MergeAppendState, fc_pstate);
	TupleTableSlot *fc_result;
	SlotNumber	fc_i;

	CHECK_FOR_INTERRUPTS();

	if (!fc_node->ms_initialized)
	{
		/* 如果所有子计划都已修剪，则无需执行任何操作 */
		if (fc_node->ms_nplans == 0)
			return ExecClearTuple(fc_node->ps.ps_ResultTupleSlot);

		/*
		 * 如果我们尚未确定有效子计划，则现在进行确认。如果
		 * 运行时修剪被禁用，则有效子计划将始终设为所有子计划。
		 */
		if (fc_node->ms_valid_subplans == NULL)
			fc_node->ms_valid_subplans =
				ExecFindMatchingSubPlans(fc_node->ms_prune_state, false);

		/*
		 * 第一次循环：从每个有效子计划中提取第一个元组，
		 * 并设置堆。
		 */
		fc_i = -1;
		while ((fc_i = bms_next_member(fc_node->ms_valid_subplans, fc_i)) >= 0)
		{
			fc_node->ms_slots[fc_i] = ExecProcNode(fc_node->mergeplans[fc_i]);
			if (!TupIsNull(fc_node->ms_slots[fc_i]))
				binaryheap_add_unordered(fc_node->ms_heap, Int32GetDatum(fc_i));
		}
		binaryheap_build(fc_node->ms_heap);
		fc_node->ms_initialized = true;
	}
	else
	{
		/*
		 * 否则，从我们上次返回的子计划中提取下一个元组，
		 * 并将子计划索引重新插入堆中，因为它现在可能与
		 * 堆中现有元素的比较结果不同。 （我们或许可以稍微简化逻辑，
		 * 在从先前的调用返回之前执行此操作，但最好是在必要时再提取元组。）
		 */
		fc_i = DatumGetInt32(binaryheap_first(fc_node->ms_heap));
		fc_node->ms_slots[fc_i] = ExecProcNode(fc_node->mergeplans[fc_i]);
		if (!TupIsNull(fc_node->ms_slots[fc_i]))
			binaryheap_replace_first(fc_node->ms_heap, Int32GetDatum(fc_i));
		else
			(void) binaryheap_remove_first(fc_node->ms_heap);
	}

	if (binaryheap_empty(fc_node->ms_heap))
	{
		/* 所有子计划都已耗尽，堆也是 */
		fc_result = ExecClearTuple(fc_node->ps.ps_ResultTupleSlot);
	}
	else
	{
		fc_i = DatumGetInt32(binaryheap_first(fc_node->ms_heap));
		fc_result = fc_node->ms_slots[fc_i];
	}

	return fc_result;
}

/*比较两个给定槽中的元组。*/
static int32 fc_heap_compare_slots(Datum fc_a, Datum fc_b, void *fc_arg)
{
	MergeAppendState *fc_node = (MergeAppendState *) fc_arg;
	SlotNumber	fc_slot1 = DatumGetInt32(fc_a);
	SlotNumber	fc_slot2 = DatumGetInt32(fc_b);

	TupleTableSlot *fc_s1 = fc_node->ms_slots[fc_slot1];
	TupleTableSlot *fc_s2 = fc_node->ms_slots[fc_slot2];
	int			fc_nkey;

	Assert(!TupIsNull(fc_s1));
	Assert(!TupIsNull(fc_s2));

	for (fc_nkey = 0; fc_nkey < fc_node->ms_nkeys; fc_nkey++)
	{
		SortSupport fc_sortKey = fc_node->ms_sortkeys + fc_nkey;
		AttrNumber	fc_attno = fc_sortKey->ssup_attno;
		Datum		fc_datum1,
					fc_datum2;
		bool		fc_isNull1,
					fc_isNull2;
		int			fc_compare;

		fc_datum1 = slot_getattr(fc_s1, fc_attno, &fc_isNull1);
		fc_datum2 = slot_getattr(fc_s2, fc_attno, &fc_isNull2);

		fc_compare = ApplySortComparator(fc_datum1, fc_isNull1,
									  fc_datum2, fc_isNull2,
									  fc_sortKey);
		if (fc_compare != 0)
		{
			INVERT_COMPARE_RESULT(fc_compare);
			return fc_compare;
		}
	}
	return 0;
}

/* ----------------------------------------------------------------
 *		ExecEndMergeAppend
 *
 *		关闭MergeAppend节点的子扫描。
 *
 *		不返回任何有趣的内容。
 * ----------------------------------------------------------------
 */
void ExecEndMergeAppend(MergeAppendState *fc_node)
{
	PlanState **fc_mergeplans;
	int			fc_nplans;
	int			fc_i;

	/*
	 * 从节点获取信息
	 */
	fc_mergeplans = fc_node->mergeplans;
	fc_nplans = fc_node->ms_nplans;

	/*
	 * 关闭每个子扫描
	 */
	for (fc_i = 0; fc_i < fc_nplans; fc_i++)
		ExecEndNode(fc_mergeplans[fc_i]);
}

void ExecReScanMergeAppend(MergeAppendState *fc_node)
{
	int			fc_i;

	/*
	 * 如果用于修剪表达式的任何 PARAM_EXEC 参数已更改，
	 * 那我们最好取消设置有效子计划，以便为新的参数值重新选择它们。
	 */
	if (fc_node->ms_prune_state &&
		bms_overlap(fc_node->ps.chgParam,
					fc_node->ms_prune_state->execparamids))
	{
		bms_free(fc_node->ms_valid_subplans);
		fc_node->ms_valid_subplans = NULL;
	}

	for (fc_i = 0; fc_i < fc_node->ms_nplans; fc_i++)
	{
		PlanState  *fc_subnode = fc_node->mergeplans[fc_i];

		/*
		 * ExecReScan 不知道我的子计划，所以我必须自己进行
		 * 变化参数的信号传递。
		 */
		if (fc_node->ps.chgParam != NULL)
			UpdateChangedParamSet(fc_subnode, fc_node->ps.chgParam);

		/*
		 * 如果子节点的 chgParam 不是 null，则计划将通过
		 * 第一个 ExecProcNode 重新扫描。
		 */
		if (fc_subnode->chgParam == NULL)
			ExecReScan(fc_subnode);
	}
	binaryheap_reset(fc_node->ms_heap);
	fc_node->ms_initialized = false;
}
